<?php
// +----------------------------------------------------------------------
// | 事件日志服务层
// +----------------------------------------------------------------------
// | 主要服务于事件日志数据处理
// +----------------------------------------------------------------------
// | 错误编码头 206xxx
// +----------------------------------------------------------------------
namespace common\service;

use common\exception\LogicException;
use common\model\EntityEpisodeRelationModel;
use common\model\EntityModel;
use common\model\NoteModel;
use common\model\ProjectMemberModel;
use common\model\ReviewFeedbackModel;
use common\model\StepUpstreamModel;
use common\model\TaskModel;
use common\model\TaskRelationTagModel;
use common\model\TaskWorkOrderModel;
use common\model\UserModel;
use common\model\VersionModel;
use common\observer\AttrChangeSendProducerAssistantMessage;
use common\observer\DesignChangeSyncToAssetLevel;
use common\observer\EntityAttributeChangeSyncToTask;
use common\observer\EntityChangeEpisodeCodeSyncToTask;
use common\observer\EntityChangeSyncToOutSourceEntity;
use common\observer\MotionSequenceTaskDoneSyncStatusToSequenceEntity;
use common\observer\OriginalPlanNodeDetailTaskClose;
use common\observer\OutsideTaskSyncDataToPrimaryProject;
use common\observer\RelationChangeAutoAssembleCheck;
use common\observer\SequenceMotionCaptorSyncToMotionTask;
use common\observer\ShotAddSequenceRelationHandle;
use common\observer\ShotDurationSyncToStoryboard;
use common\observer\StartEndTimeAutoSave;
use common\observer\StatusSyncToParentStepCategoryTask;
use common\observer\StepUpstreamUpdateRemoveCache;
use common\observer\StoryboardAttrSyncToShot;
use common\observer\task\version\InsideVersionPassRecordEndTime;
use common\observer\task\version\VersionPublishNotifyRelateTask;
use common\observer\task\version\VersionUploadedRecordUploadedVersion;
use common\observer\task\version\OutsideVersionUpdateSyncToInsideVersion;
use common\observer\task\WorkOrder\StatusChangeSendMessageToChat;
use common\observer\TaskAssignInviteChatUser;
use common\observer\TaskCheckPlanTimeIsOverdue;
use common\observer\TaskDoneCheckOrderSettlementStatus;
use common\observer\TaskDoneDownstreamAssembleCheck;
use common\observer\TaskDoneNotifyRelateTask;
use common\observer\TaskEpisodeIdSyncTaskRelationTag;
use common\observer\TaskExecutorChangeNotifyUpstreamDoneStatus;
use common\observer\TaskExecutorChangeSyncTaskMemberStatus;
use common\observer\TaskExecutorTypeSyncToEntity;
use common\observer\TaskPlanTimeUpdate;
use common\observer\TaskReadyToPublishPushTenantEvent;
use common\observer\TaskStatusChangeNotifyExecutor;
use common\observer\TaskStatusSyncToEntity;
use common\observer\TaskStatusSyncToEpisode;
use common\observer\TaskStatusSyncToTaskMember;
use common\observer\TaskStepCategoryPlanEndTime;
use common\observer\TexStepTodoTaskMemberExecutorSync;
use common\observer\UpdateEntitySendMessageToChat;
use common\observer\UpdateManhour;
use support\bootstrap\Log;
use support\ErrorCode;
use support\SingletonTrait;
use think\Exception;
use think\Hook;
use think\module\Module;
use Throwable;
use Webman\Stomp\Client;
use function request;

class EventLogService
{
    // 单例
    use SingletonTrait;

    // 错误信息
    protected $errorMsg = '';

    /**
     * 获取错误信息
     * @return string
     */
    public function getError()
    {
        return $this->errorMsg;
    }

    /**
     * 记录到Event服务器
     * @param $data
     * @param $controllerMethod
     * @return array
     */
    protected function postToServer($data, $controllerMethod)
    {
        return teamones_request('teamones-log', $controllerMethod, $data);
    }

    /**
     * 记录事件日志
     * @param $param
     */
    public function create($param)
    {
        //--调试--Eventlog 日志
        $now = time();
        if ($now > strtotime('2023/11/27 15:00:00') && $now< strtotime('2023/11/27 15:30:00')) {
            trace('event_log'.json_encode($param), 'DEBUG');
        }
        if ($now > strtotime('2023/11/28 03:00:00') && $now< strtotime('2023/11/28 04:00:00')) {
            trace('event_log'.json_encode($param), 'DEBUG');
        }
        //--调试--Eventlog 日志

        $updateEvents = array_filter($param, function ($item) {
            return isset($item['module_code']) && isset($item['operate']) && $item['operate'] == 'update';
        });
        $updateEvents = array_group_by($updateEvents, 'module_code');

        foreach ($updateEvents as $moduleCode => $eventsGrouped) {
            switch ($moduleCode) {
                case 'task':
                    // 写入到log-server
//                    try {
//                        $writeLogData = array_values($eventsGrouped);
//                        $this->postToServer(['data' => $writeLogData], "log/add");
//                    } catch (Throwable $e) {
//                        trace("postToServer got exception", 'ERR');
//                        trace(format_exception_info($e), 'ERR');
//                    }
                    // 处理任务状态
                    try {
                        $this->handleTaskStatusUpdate($eventsGrouped);
                    } catch (Throwable $e) {
                        trace("handleTaskStatusUpdate got exception", 'ERR');
                        trace(format_exception_info($e), 'ERR');
                    }
                    break;
                case 'entity':
                    try {
                        $this->handleEntityUpdateEvents($eventsGrouped);
                    } catch (Throwable $e) {
                        trace("[处理实体update事件] got exception", 'ERR');
                        trace(format_exception_info($e), 'ERR');
                    }
                    break;
                case 'version':
                    try {
                        $this->handleVersionUpdateEvents($eventsGrouped);
                    } catch (Throwable $e) {
                        trace('handleVersionEvents got error', 'ERR');
                        trace(format_exception_info($e), 'ERR');
                    }
                    break;
                case 'note':
                    try {
                        $this->handleNoteUpdateEvents($eventsGrouped);
                    } catch (Throwable $e) {
                        trace('handleNoteUpdateEvents got error', 'ERR');
                        trace(format_exception_info($e), 'ERR');
                    }
                    break;
                case 'step_upstream':
                    try {
                        $this->handleStepUpstreamUpdateEvents($eventsGrouped);
                    } catch (Throwable $e) {
                        trace('handleStepUpstreamUpdateEvents got error', 'ERR');
                        trace(format_exception_info($e), 'ERR');
                    }
                    break;
                case 'review_feedback':
                    try {
                        $this->handleUpdateOriginalPlanNodeEvents($eventsGrouped);
                        $this->handleReviewFeedbackUpdateEvents($eventsGrouped);
                    } catch (Throwable $e) {
                        trace('handleNoteUpdateEvents got error', 'ERR');
                        trace(format_exception_info($e), 'ERR');
                    }
                    break;
                case 'task_work_order':
                    try {
                        $this->handleTaskWorkOrderUpdateEvents($eventsGrouped);
                    } catch (Throwable $e) {
                        trace('handleTaskWorkOrderUpdateEvents got error', 'ERR');
                        trace(format_exception_info($e), 'ERR');
                    }
                    break;
                case 'delay_order':
                    try {
                        $this->handleUpdateOriginalPlanNodeEvents($eventsGrouped);
                    } catch (Throwable $e) {
                        trace('handleUpdateOriginalPlanNodeEvents got error', 'ERR');
                        trace(format_exception_info($e), 'ERR');
                    }
                    break;
                case 'plan':
                    $this->handleUpdateOriginalPlanNodeEvents($eventsGrouped);
//                    try {
//                        // 排期修改---写入到log-server
//                        $writeLogData = array_values($eventsGrouped);
//                        $this->postToServer(['data' => $writeLogData], "log/add");
//
//                    } catch (Throwable $e) {
//                        trace('handleUpdateOriginalPlanNodeEvents got error', 'ERR');
//                        trace(format_exception_info($e), 'ERR');
//                    }
                    break;
            }
        }


        // 填充初始集数
//        try {
//            $this->fillInitialEpisodeCode($param);
//        } catch (Throwable $e) {
//            trace("fillInitialEpisodeCode got exception", 'ERR');
//            trace(format_exception_info($e), 'ERR');
//        }

        // 处理项目创建事件
        try {
            $projectCreateEvents = array_filter($param, function ($item) {
                return isset($item['module_code']) && $item['module_code'] == 'project' && isset($item['operate']) && $item['operate'] == 'create';
            });
            if (!empty($projectCreateEvents)) {
                $this->handleProjectCreate($projectCreateEvents);
            }
        } catch (Throwable $e) {
            trace("handleProjectCreate got exception", 'ERR');
            trace(format_exception_info($e), 'ERR');
        }

        // 处理项目成员管理事件
        try {
            $projectMemberChangeEvents = array_filter($param, function ($item) {
                return isset($item['module_code']) && $item['module_code'] == 'project_member' && isset($item['operate']) && in_array($item['operate'], ['create', 'delete']);
            });
            if (!empty($projectMemberChangeEvents)) {
                $this->handleProjectMemberChange($projectMemberChangeEvents);
            }

        } catch (Throwable $e) {
            trace("handleProjectMemberChange got exception", 'ERR');
            trace(format_exception_info($e), 'ERR');
        }


        try {
            $entityCreateEvents = array_filter($param, function ($item) {
                return isset($item['module_code']) && $item['module_code'] === 'entity' && isset($item['operate']) && in_array($item['operate'], ['create', 'update', 'delete']);
            });
            if (!empty($entityCreateEvents)) {
                $this->handleEntityEpisodeRelationEvents($entityCreateEvents);
            }
        } catch (Throwable $e) {
            trace("[处理维护实体与集关联表] got exception", 'ERR');
            trace(format_exception_info($e), 'ERR');
        }

        $createEvents = array_filter($param, function ($item) {
            return isset($item['module_code']) && isset($item['operate']) && $item['operate'] == 'create';
        });
        $createEventsGrouped = array_group_by($createEvents, 'module_code');

        foreach ($createEventsGrouped as $moduleCode => $createEvents) {
            try {
                switch ($moduleCode) {
                    case 'note':
                        $this->handleNoteCreateEvents($createEvents);
                        break;
                    case 'task':
                        $this->handleTaskCreatedEvents($createEvents);
                        break;
                    case 'step_upstream':
                        $this->handleStepUpstreamCreateEvents($createEvents);
                        break;
                    case 'version':
                        $this->handleVersionCreateEvents($createEvents);
                        break;
                    case 'review_request':
                        $this->handleReviewRequestEvents($createEvents);
                        break;
                    case 'review_feedback':
                        $this->handleUpdateOriginalPlanNodeEvents($createEvents);
                        $this->handleReviewCreateEvents($createEvents);
                        break;
                    case 'plan':
                        $this->handleUpdateOriginalPlanNodeEvents($createEvents);
                        break;
                }
            } catch (Throwable $e) {
                trace(format_exception_info($e), 'ERR');
            }
        }
        //删除事件
        $deleteEvents = array_filter($param, function ($item) {
            return isset($item['module_code']) && isset($item['operate']) && $item['operate'] == 'delete';
        });
        $deleteEventsGrouped = array_group_by($deleteEvents, 'module_code');

        foreach ($deleteEventsGrouped as $moduleCode => $deleteEvents) {
            try {
                switch ($moduleCode) {
                    case 'plan':
                        $this->handleUpdateOriginalPlanNodeEvents($deleteEvents);
                        break;
                }
            } catch (Throwable $e) {
                trace(format_exception_info($e), 'ERR');
            }
        }

    }

    /**
     * 写入交易记录日志
     * @param $param
     * @param $entryDirection
     */
    public function writeTransactionRecords($param, $entryDirection = 0)
    {
        /**
         * 字段说明
         * type: recharge（充值）、order（结算）
         * entry_direction：0 接收订单，1 结算
         * amount：金额
         * link_id：订单id
         * created_by：操作者uuid
         * belong_system：所属系统
         */

        $userUUID = request()->getUserUUID();
        $param = [
            "type" => "order",
            "entry_direction" => $entryDirection,
            "amount" => (float)$param['money'],
            "link_id" => (int)$param['id'],
            "created_by" => (string)$userUUID,
            "belong_system" => (string)C('belong_system')
        ];


        // 记录到Event服务器
//        try {
//            $this->postToServer($param, "transaction_records/add");
//        } catch (Throwable $e) {
//            Log::error("postToServer got exception");
//            Log::error("##########################");
//            Log::error($e->getMessage());
//            Log::error($e->getCode());
//            Log::error($e->getTraceAsString());
//            Log::error("##########################");
//        }
    }

    /**
     * 填充初始集数
     * @param $param
     * @return void
     * @throws \Exception
     */
    public function fillInitialEpisodeCode($param)
    {
        foreach ($param as $value) {
            if ($value['module_code'] == 'entity' && in_array($value['operate'], ['create', 'update'])) {
                $old = json_decode($value['record']['data']['old']['json'] ?? "", true);
                $new = json_decode($value['record']['data']['new']['json'] ?? "", true);
                $oldAsset = (isset($old['sequence_asset']) && !empty($old['sequence_asset'])) ? explode(',', $old['sequence_asset']) : [];
                $newAsset = (isset($new['sequence_asset']) && !empty($new['sequence_asset'])) ? explode(',', $new['sequence_asset']) : [];
                if (!empty($oldAsset) || !empty($newAsset)) {
                    $this->updateEpisodeCode($value['link_id'], $oldAsset, $newAsset);
                }
                $oldLevel = (isset($old['sequence_level']) && !empty($old['sequence_level'])) ? explode(',', $old['sequence_level']) : [];
                $newLevel = (isset($new['sequence_level']) && !empty($new['sequence_level'])) ? explode(',', $new['sequence_level']) : [];
                if (!empty($oldLevel) || !empty($newLevel)) {
                    $this->updateEpisodeCode($value['link_id'], $oldLevel, $newLevel);
                }
                $oldLevelAsset = (isset($old['level_asset']) && !empty($old['level_asset'])) ? explode(',', $old['level_asset']) : [];
                $newLevelAsset = (isset($new['level_asset']) && !empty($new['level_asset'])) ? explode(',', $new['level_asset']) : [];
                if (!empty($oldLevelAsset) || !empty($newLevelAsset)) {
                    $this->updateEpisodeCode($value['link_id'], $oldLevelAsset, $newLevelAsset);
                }
            }
        }
    }

    /**
     * 更新资产或关卡的初始集
     * @param $sequence_id
     * @param $old
     * @param $new
     * @throws \Exception
     */
    protected function updateEpisodeCode($sequence_id, $old, $new)
    {
        $entityModel = model(EntityModel::class);
        $sequence = $entityModel->find($sequence_id);
        if (!empty($sequence) && !empty($sequence['initial_episode_code'])) {
            if (count($old) < count($new)) {
                //关联
                $diff = array_diff($new, $old);
                $assets = $entityModel->select(implode(',', $diff));
                $updateArr = [];
                foreach ($assets as $value) {
                    if (empty($value['initial_episode_code'])) {
                        //如果资产不存在初始集直接更新
                        $updateArr[] = $value['id'];
                    } else {
                        //存在初始集，决定是否更新初始集
                        if (strnatcmp($value['initial_episode_code'], $sequence['initial_episode_code']) > 0) {
                            $updateArr[] = $value['id'];
                        }
                    }
                }
                if (!empty($updateArr)) {
                    $entityModel->where(['id' => ['IN', join(',', $updateArr)]])->setField('initial_episode_code', $sequence['initial_episode_code']);
                }
            } else {
                //取消关联
                $diff = array_diff($old, $new);
                $assets = $entityModel->select(implode(',', $diff));
                foreach ($assets as $value) {
                    //查看资产是否被其它序列关联
                    $initial_episode_code = $entityModel->where("FIND_IN_SET(" . $value['id'] . ",JSON_UNQUOTE(JSON_EXTRACT(json,'$.sequence_asset'))) OR FIND_IN_SET(" . $value['id'] . ",JSON_UNQUOTE(JSON_EXTRACT(json,'$.sequence_level')))")->order("LENGTH(initial_episode_code),initial_episode_code ASC")->limit(1)->getField("initial_episode_code");
                    $entityModel->where(['id' => $value['id']])->setField('initial_episode_code', !empty($initial_episode_code) ? $initial_episode_code : '');
                }
            }
        }
    }

    /**
     * 日志查询
     * @param $params
     * @return array
     */
    public function select($params)
    {
        $data = $this->postToServer($params, "log/select");
        if ($data === false) {
            return [];
        }
        return $data;
    }


    /**
     * 获得格式化的日志数据
     * @param $param
     * @return array
     */
    public function getEventLogFormatted($param)
    {
        try {
            $filter = [
                'module_code' => $param['filter']['module_code'],
                'link_id' => $param['filter']['link_id'],
                'belong_system' => C("belong_system"),
            ];
            $page = $param['page'] ?? [1, C("database.database_max_select_rows")];
            $order = $param['order'] ?? "id,desc";

            $requestData = [
                "param" => [
                    "fields" => "id,level,operate,created_by,batch_number,belong_system,link_id,module_code,record,created,uuid",
                    "filter" => $filter,
                    "page" => $page,
                    "order" => $order
                ]
            ];
            $res = $this->select($requestData);

            // 获取出当前系统的用户uuid
            $userUUIDS = array_column($res['rows'], 'created_by', 'created_by');
            $usersMapByUUID = [];
            if (!empty($userUUIDS)) {
                $userModel = model(UserModel::class);
                $users = $userModel
                    ->where(['uuid' => ['IN', join(',', $userUUIDS)]])
                    ->field("id,name,email,uuid")
                    ->select();
                $usersMapByUUID = array_column($users, null, 'uuid');
            }

            $this->formatTaskEventLog($res['rows'], $usersMapByUUID, $param['filter']['module_code']);
            return $res;
        } catch (Throwable $e) {
            throw new LogicException($e->getMessage(), ErrorCode::FORMAT_EVENT_LOG_FAILURE);
        }
    }

    /**
     * 格式化事件日志
     * @param $logs
     * @param $usersMapByUUID
     * @param $moduleCode
     */
    private function formatTaskEventLog(&$logs, $usersMapByUUID, $moduleCode)
    {
        $moduleName = Module::$moduleDictData['module_index_by_code'][$moduleCode]['name'];

        //处理一些需要关联查询的数据
        $userIds = [];
        foreach ($logs as &$log) {
            switch ($log['operate']) {
                case "update":
                    $new = $log['record']['data']['new'] ?? [];
                    $old = $log['record']['data']['old'] ?? [];
                    $diffAttr = array_diff_assoc($new, $old);
                    foreach ($diffAttr as $changedKey => $changedValue) {
                        //筛选出需要查询的用户
                        if (in_array($changedKey, ['executor', 'assignee']) && (int)$changedValue > 0) {
                            $userIds[] = $changedValue;
                        }
                    }
                    break;
                default:
            }
        }

        $userMapById = [];
        //追加额外的用户信息 assignee executor信息
        if (!empty($userIds)) {
            $userModel = model(UserModel::class);
            $usersExtra = $userModel
                ->where(['id' => ['IN', join(',', $userIds)]])
                ->field("id,name,email")
                ->select();
            $userMapById = array_column($usersExtra, null, 'id');
        }

        foreach ($logs as &$log) {
            $log['created_by'] = $usersMapByUUID[$log['created_by']] ?? null;
            $changes = [];
            switch ($log['operate']) {
                case "update":
                    $new = $log['record']['data']['new'] ?? [];
                    $old = $log['record']['data']['old'] ?? [];
                    $diffAttr = array_diff_assoc($new, $old);
                    foreach ($diffAttr as $changedKey => $changedValue) {
                        //如果是json字段 拆开进行解析对比
                        if (in_array($changedKey, ['config', 'json'])) {
                            $oldJsonData = json_decode($old[$changedKey], true) ?? [];
                            $newJsonData = json_decode($new[$changedKey], true) ?? [];
                            $diffJsonAttr = array_diff_assoc($newJsonData, $oldJsonData);
                            foreach ($diffJsonAttr as $diffJsonKey => $diffJsonValue) {
                                $changes[] = ['field' => $diffJsonKey, 'value' => $diffJsonValue];
                            }
                            continue;
                        }

                        //如果是修改的执行人 分配人属性
                        if (in_array($changedKey, ['executor', 'assignee'])) {
                            if (empty($userMapById[$changedValue])) {
                                $changes[] = ['field' => $changedKey, "value" => "空"];
                            } else {
                                $changes[] = ['field' => $changedKey, "value" => $userMapById[$changedValue]['name']];
                            }
                            continue;
                        }
                        //普通属性修改
                        $changes[] = ['field' => $changedKey, 'value' => $changedValue];
                    }

                    break;
                case "create":
                case "delete":
                    $changes = [];
                    break;
                default:
                    $changes = "{$log['operate']}了{$moduleName}";
            }
            $log['changes'] = $changes;
            unset($log['record'], $log['belong_system'], $log['batch_number'], $log['uuid']);
        }
    }

    /**
     * 处理任务状态更新监听
     * 同步工序任务和工序分类任务的状态
     *  工序任务开始  分类任务开始
     *  工序任务全部完成  分类任务完成
     * @param $taskEvents
     * @throws \Exception
     */
    private function handleTaskStatusUpdate($taskEvents)
    {
        $taskObServerSubject = observer_subject_by_class_name(TaskModel::class);
        $taskObServerSubject->attach(TexStepTodoTaskMemberExecutorSync::getInstance());
        $taskObServerSubject->attach(TaskStatusChangeNotifyExecutor::getInstance());
        $taskObServerSubject->attach(StatusSyncToParentStepCategoryTask::getInstance());
        $taskObServerSubject->attach(OutsideTaskSyncDataToPrimaryProject::getInstance());
        $taskObServerSubject->attach(StartEndTimeAutoSave::getInstance());
        $taskObServerSubject->attach(TaskPlanTimeUpdate::getInstance());
        $taskObServerSubject->attach(TaskDoneDownstreamAssembleCheck::getInstance());
        $taskObServerSubject->attach(MotionSequenceTaskDoneSyncStatusToSequenceEntity::getInstance());
        $taskObServerSubject->attach(TaskReadyToPublishPushTenantEvent::getInstance());
        $taskObServerSubject->attach(TaskStatusSyncToEntity::getInstance());
        $taskObServerSubject->attach(TaskExecutorTypeSyncToEntity::getInstance());
        $taskObServerSubject->attach(TaskAssignInviteChatUser::getInstance());
        $taskObServerSubject->attach(TaskExecutorChangeNotifyUpstreamDoneStatus::getInstance());
        $taskObServerSubject->attach(TaskDoneCheckOrderSettlementStatus::getInstance());
        $taskObServerSubject->attach(TaskDoneNotifyRelateTask::getInstance());
        $taskObServerSubject->attach(TaskStepCategoryPlanEndTime::getInstance());
        $taskObServerSubject->attach(TaskCheckPlanTimeIsOverdue::getInstance());
        $taskObServerSubject->attach(TaskStatusSyncToEpisode::getInstance());
        $taskObServerSubject->attach(TaskStatusSyncToTaskMember::getInstance());
        $taskObServerSubject->attach(TaskExecutorChangeSyncTaskMemberStatus::getInstance());
        $taskObServerSubject->attach(TaskEpisodeIdSyncTaskRelationTag::getInstance());
        $taskObServerSubject->attach(OriginalPlanNodeDetailTaskClose::getInstance());

        $tasks = model(TaskModel::class)
            ->where(['id' => ['IN', implode(',', array_column($taskEvents, 'link_id'))]])
            ->select();
        $tasks = array_column($tasks, null, 'id');
        foreach ($taskEvents as $event) {
            try {
                $id = $event['link_id'];
                $taskObServerSubject->setSplPrimaryId($id);
                //空判断
                $subjectData = $tasks[$id] ?? [];
                $taskObServerSubject->setSubjectData($subjectData);
                $changed = $taskObServerSubject->splDataChanged($event['record']['data']['old'], $event['record']['data']['new']);
                trace('task attr changed:' . PHP_EOL . var_export($changed, true), 'DEBUG');
                $taskObServerSubject->notify();
            } catch (Throwable $e) {
                trace(format_exception_info($e), 'ERR');
            }
        }
        APP_DEBUG && trace(__METHOD__ . " PROCESS DONE", "DEBUG");
    }

    /**
     * 处理项目创建事件
     * @param array $projectCreateEvents
     */
    private function handleProjectCreate(array $projectCreateEvents)
    {
        foreach ($projectCreateEvents as $event) {
            $data = $event['record']['data'];
//            if (isset($data['type']) && $data['type'] == "outsource") {
//                continue;
//            }

            $chat = [
                'name' => $data['name'],
                'type' => 'group',
                'project_id' => $data['id'],
                'link_id' => $data['id'],
                'link_type' => 'project',
                'from_tenant_id' => $data['tenant_id'],
                'to_tenant_id' => 0
            ];
            try {
                $requestData = ['data' => [
                    'chat' => $chat,
                    'chat_users' => [],
                    'owner' => 0
                ]];
                teamones_request('teamones-im', 'oauth/create_chat', $requestData);
            } catch (Throwable $e) {
                Log::error("##########################");
                Log::error($e->getMessage());
                Log::error($e->getCode());
                Log::error($e->getTraceAsString());
                Log::error("##########################");
            }
        }
        APP_DEBUG && trace(__METHOD__ . " PROCESS DONE", "DEBUG");
    }

    /**
     * 处理项目成员管理变动
     * @param array $projectMemberChangeEvents
     */
    private function handleProjectMemberChange(array $projectMemberChangeEvents)
    {
        foreach ($projectMemberChangeEvents as $event) {
            try {
                if ($event['operate'] === "create") {
                    // 将用户添加进项目群
                    $data = $event['record']['data'];
                    $requestData = ['param' => ['filter' => [
                        'link_type' => "project",
                        'link_id' => $data['project_id']
                    ]]];
                    $projectChat = teamones_request('teamones-im', 'oauth/find_chat', $requestData)['data'];
                    if (empty($projectChat)) {
                        //没有项目群
                        APP_DEBUG && trace("do not have a project chat", "DEBUG");
                        continue;
                    }
                    $unionId = model(UserModel::class)->field('union_id')->where(['id' => $data['user_id']])->value('union_id');
                    if (empty($unionId)) {
                        continue;
                    }
                    $requestData = ['data' => [
                        'chat_global_id' => $projectChat['global_id'],
                        'user_id' => [$unionId],
                    ]];
                    APP_DEBUG && trace("添加用户到项目群聊 $unionId", "DEBUG");
                    APP_DEBUG && trace(var_export($requestData, true), "DEBUG");
                    teamones_request('teamones-im', 'oauth/invite_chat_user', $requestData);
                    continue;
                } else if ($event['operate'] === "delete") {
                    $records = array_values($event['record']['data']);
                    $projectId = $records[0]['project_id'];
                    // 将用户移出项目群
                    $userIdData = array_column($records, 'user_id');

                    // 检测用户是否还在项目
                    $existUserIdList = model(ProjectMemberModel::class)->where(['project_id' => $projectId, 'user_id' => ['IN', implode(',', $userIdData)]])->field('user_id')->column('user_id');

                    $realDeleteUserIdList = [];
                    foreach ($userIdData as $userId) {
                        if (in_array($userId, $existUserIdList)) {
                            continue;
                        }
                        $realDeleteUserIdList[] = $userId;
                    }
                    if (empty($realDeleteUserIdList)) {
                        continue;
                    }

                    $unionIdList = model(UserModel::class)
                        ->where([
                            'id' => ['IN', implode(',', $realDeleteUserIdList)],
                        ])
                        ->field('union_id')
                        ->column('union_id');

                    if (empty($unionIdList)) {
                        continue;
                    }
                    $unionIds = implode(',', $unionIdList);

                    $requestData = ['param' => ['filter' => [
                        'link_type' => "project",
                        'link_id' => $projectId
                    ]]];
                    $projectChat = teamones_request('teamones-im', 'oauth/find_chat', $requestData)['data'];
                    if (empty($projectChat)) {
                        //没有项目群
                        APP_DEBUG && trace("do not have a project chat", "DEBUG");
                        continue;
                    }
                    // 删除项目聊天用户
                    $requestData = ['param' => ['filter' => [
                        'user_id' => [
                            "-in",
                            $unionIds
                        ],
                        'chat_global_id' => $projectChat['global_id'],
                    ]]];
                    teamones_request('teamones-im', 'oauth/delete_chat_user', $requestData);
                    APP_DEBUG && trace("从项目会话中移除 $unionIds", "DEBUG");
                    APP_DEBUG && trace(var_export($requestData, true), "DEBUG");
                    continue;
                }
            } catch (Throwable $e) {
                trace(format_exception_info($e), 'ERR');
            }
        }
        APP_DEBUG && trace(__METHOD__ . " PROCESS DONE", "DEBUG");
    }

    /**
     * 处理实体更新事件
     *  更新预估工时
     * @param array $entityUpdateEvents
     * @throws \Exception
     */
    private function handleEntityUpdateEvents(array $entityUpdateEvents)
    {
        $entityObServerSubject = observer_subject_by_class_name(EntityModel::class);
        $entityObServerSubject->attach(SequenceMotionCaptorSyncToMotionTask::getInstance());
        $entityObServerSubject->attach(UpdateManhour::getInstance());
        $entityObServerSubject->attach(StoryboardAttrSyncToShot::getInstance());
        $entityObServerSubject->attach(DesignChangeSyncToAssetLevel::getInstance());
        $entityObServerSubject->attach(RelationChangeAutoAssembleCheck::getInstance());
        $entityObServerSubject->attach(UpdateEntitySendMessageToChat::getInstance());
        $entityObServerSubject->attach(EntityChangeSyncToOutSourceEntity::getInstance());
        $entityObServerSubject->attach(AttrChangeSendProducerAssistantMessage::getInstance());
        $entityObServerSubject->attach(ShotAddSequenceRelationHandle::getInstance());
        $entityObServerSubject->attach(ShotDurationSyncToStoryboard::getInstance());
        $entityObServerSubject->attach(EntityChangeEpisodeCodeSyncToTask::getInstance());
        $entityObServerSubject->attach(EntityAttributeChangeSyncToTask::getInstance());

        $entityFilter = ['id' => ['IN', array_column($entityUpdateEvents, 'link_id')]];
        $entityList = model(EntityModel::class)->where($entityFilter)->select();
        $entityList = array_column($entityList, null, 'id');
        foreach ($entityUpdateEvents as $event) {
            try {
                // 监听属性改变 同时观察者
                $id = $event['link_id'];
                $entityObServerSubject->splDataChanged($event['record']['data']['old'], $event['record']['data']['new']);
                $entityObServerSubject->setSplPrimaryId($id);
                //空判断
                $subjectData = $entityList[$id] ?? [];
                $entityObServerSubject->setSubjectData($subjectData);
                $entityObServerSubject->notify();
            } catch (Throwable $e) {
                trace(format_exception_info($e), 'ERR');
            }
        }
    }

    /**
     * 通过实体curd事件 维护集与实体的关系
     * @param array $entityCreateEvents
     * @throws \Exception
     */
    private function handleEntityEpisodeRelationEvents(array $entityCreateEvents)
    {
        foreach ($entityCreateEvents as $event) {
            APP_DEBUG && trace(__METHOD__ . ":" . $event['operate'], 'DEBUG');
            try {
                switch ($event['operate']) {
                    case 'create':
                        $data = $event['record']['data'];
                        //维护实体与集关系表
                        if (isset($data['initial_episode_code'])) {
                            $episode = model(EntityModel::class)
                                ->where(['project_id' => $data['project_id'], 'module_id' => Module::$moduleDictData['module_index_by_code']['episode']['id'], 'code' => $data['initial_episode_code']])->find();
                            if (!empty($episode)) {
                                $addData = [
                                    'entity_id' => $data['id'],
                                    'entity_module_id' => $data['module_id'],
                                    'project_id' => $data['project_id'],
                                    'episode_id' => $episode['id'],
                                    'showings_number' => $data['showings_number'] ?? ''
                                ];
                                model(EntityEpisodeRelationModel::class)->addItem($addData);
                                APP_DEBUG && trace("添加实体集关系 {$data['initial_episode_code']}-{$addData['showings_number']}-{$data['name']}", 'DEBUG');
                                //反填入剧本拆分的场次
                                $this->fillBackScriptList($data, $episode);

                                try {
                                    $this->entityCreateRelationEntityAddEpisodeRelation($data, $episode);
                                } catch (Throwable $e) {
                                    trace(format_exception_info($e), 'ERR');
                                }
                            }
                            $entity = model(EntityModel::class)->find($data['id']);
                            if (!empty($entity)) {
                                ChatService::getInstance()->createEntityChat($data);
                            }
                        }
                        //计算实体预估工时
                        Client::send('calculate_entity_man_hour', build_queue_data_with_xu(['id' => $data['id']]), 10);

                        break;
                    case 'update':
                        $record = $event['record'];
                        try {
                            $this->updateEntityEpisodeRelation($record);
                        } catch (Throwable $e) {
                            trace("更新实体集关系" . format_exception_info($e), 'ERR');
                        }

                        break;
                    case 'delete':
                        $data = $event['record']['data'];
                        foreach ($data as $dataItem) {
                            model(EntityEpisodeRelationModel::class)->where(['entity_id' => $dataItem['id']])->delete();
                            trace("删除实体集关系 {$dataItem['id']}", 'DEBUG');
                            //删除掉序列分镜的关联关系
                            $this->deleteSequenceStoryboardRelations($dataItem);
                        }
                        break;
                }

                // 同步概念的集实体关系 到资产
                $entity = model(EntityModel::class)->field('id,module_id,project_id')->find($event['record']['primary_id']);
                if (!empty($entity) && $entity['module_id'] === Module::$moduleDictData['module_index_by_code']['design']['id']) {
                    Client::send('sync_design_episode_relation_to_asset', build_queue_data_with_xu(['filter' => ['project_id' => $entity['project_id'], 'id' => $entity['id']]]));
                }
            } catch (Throwable $e) {
                trace(format_exception_info($e), 'ERR');
            }
        }
    }

    /**
     * 删除掉序列分镜的关联关系
     * @param $data
     */
    public function deleteSequenceStoryboardRelations($data)
    {
        trace("开始删除删除掉序列分镜的关联关系 {$data['id']}", 'DEBUG');
        if (empty($data)) {
            return;
        }

        if ($data['module_id'] === Module::$moduleDictData['module_index_by_code']['sequence']['id']) {
            $storyboard = model(EntityModel::class)
                ->where(["project_id" => $data['project_id'], "module_id" => module_by_code('storyboard')['id'],
                    "JSON_UNQUOTE(JSON_EXTRACT(json, '$.storyboard_motion_sequence'))" => $data['id']])
                ->field("id,JSON_UNQUOTE(JSON_EXTRACT(json, '$.storyboard_motion_sequence')) as storyboard_motion_sequence")
                ->find();

            if (!empty($storyboard["storyboard_motion_sequence"])) {
                model(EntityModel::class)->execute("UPDATE entity SET json=json_set(`json`,'$.storyboard_motion_sequence','') WHERE id = {$storyboard['id']}");
                trace("清空分镜{$storyboard['id']}的序列关联关系 {$data['id']}", 'DEBUG');
            }
        }
    }

    /**
     * 更新实体与集关联关系
     * @param $record
     * @throws \Exception
     */
    public function updateEntityEpisodeRelation($record)
    {
        $old = $record['data']['old'];
        $new = $record['data']['new'];

        $entityData = model(EntityModel::class)->find($record['primary_id']);

        //集不能未空, 且集场都未更新,则直接结束
        if (isset($new['initial_episode_code'])) {
            if (empty($new['initial_episode_code'])) {
                return;
            }
        } elseif (!isset($new['showings_number'])) {
            return;
        }

        //1 . 删除掉旧的   -实体与集场关联关系
        //获取旧集的数据
        if (isset($old['initial_episode_code'])) {
            $oldInitialEpisodeCode = $old['initial_episode_code'];
        } else { //否则获取旧的集
            $oldInitialEpisodeCode = $entityData['initial_episode_code'];
        }

        //获取旧场的数据
        if (isset($old['showings_number'])) {
            $oldShowingsNumber = $old['showings_number'];
        } else { //否则获取旧的集
            $oldShowingsNumber = $entityData['showings_number'];
        }

        $episodeModuleId = Module::$moduleDictData['module_index_by_code']['episode']['id'];
        //获取旧集id
        $oldEpisode = model(EntityModel::class)->where(['project_id' => $entityData['project_id'], 'module_id' => $episodeModuleId, 'code' => $oldInitialEpisodeCode])->field('id')->find();

        $entityEpisodeRelationData = [
            'entity_id' => $entityData['id'],
            'entity_module_id' => $entityData['module_id'],
            'project_id' => $entityData['project_id'],
        ];

        //判断是否集更新
        if (isset($new['initial_episode_code'])) {
            $newInitialEpisodeCode = $new['initial_episode_code'];
        } else {
            if (empty($entityData['initial_episode_code'])) {  //不能删除旧关联关系
                return;
            }
            $newInitialEpisodeCode = $entityData['initial_episode_code'];
        }

        if (!empty($oldEpisode)) {
            $oldEpisodeId = $oldEpisode['id'];
            //清楚掉旧数据的集与场
            $entityEpisodeRelationData['episode_id'] = $oldEpisodeId;
            $entityEpisodeRelationData['showings_number'] = $oldShowingsNumber;
            model(EntityEpisodeRelationModel::class)->where($entityEpisodeRelationData)->delete();
            trace("删除实体集关系 {$oldEpisodeId}-{$oldShowingsNumber}-{$entityData['name']}", 'DEBUG');
        }

        //旧的空集不会入录 直接入录新的
        $newEpisode = model(EntityModel::class)->where(['project_id' => $entityData['project_id'], 'module_id' => $episodeModuleId, 'code' => $newInitialEpisodeCode])->field('id')->find();

        if (empty($newEpisode)) {
            return;
        }

        //判断是否场更新
        if (isset($new['showings_number'])) {
            $newShowingsNumber = $new['showings_number'];
        } else { //否则获取旧的场
            $newShowingsNumber = $entityData['showings_number'];
        }

        //添加新数据的集与场
        $entityEpisodeRelationData['episode_id'] = $newEpisode['id'];
        $entityEpisodeRelationData['showings_number'] = $newShowingsNumber;

        model(EntityEpisodeRelationModel::class)->addItem($entityEpisodeRelationData);
        trace("添加实体集关系 {$newEpisode['id']}-{$newShowingsNumber}-{$entityData['name']}", 'DEBUG');


        //反同步填入剧本拆分的场次
        if (!empty($entityEpisodeRelationData['showings_number'])) {
            $sceneData = [
                [
                    'project_id' => $entityEpisodeRelationData['project_id'],
                    "episode_id" => $entityEpisodeRelationData['episode_id'],
                    "showings_number" => $entityEpisodeRelationData['showings_number'],
                ]
            ];
            $entityService = EntityService::getInstance();
            $entityService->fillBackScriptSceneData($sceneData);
        }
    }

    /**
     * 回填清单数据
     * @param $data
     * @param $episode
     * @return array
     * @throws \Exception
     */
    public function fillBackScriptList($data, $episode)
    {
        if (empty($episode) || empty($data['showings_number'])) {
            return [];
        }

        $sceneData = [
            [
                'project_id' => $data['project_id'],
                "episode_id" => $episode['id'],
                "showings_number" => $data['showings_number'],
            ]
        ];

        //同步集的数据
        $entityService = EntityService::getInstance();
        $entityService->fillBackScriptSceneData($sceneData);

        //同步概念清单的数据到剧本
        if (!empty($data['module_id']) && Module::$moduleDictData['module_index_by_id'][$data['module_id']]['code'] === "design") {
            $designCategory = "";
            $designPerformSpecies = "";
            if (!empty($data['json'])) {
                $json = json_decode($data['json'], true);
                $designCategory = $json['design_category'] ?? "";
                $designPerformSpecies = $json['design_perform_species'] ?? "";
            }

            $detailedListData = [
                [
                    'project_id' => $data['project_id'],
                    "episode_id" => $episode['id'],
                    "showings_number" => $data['showings_number'],
                    "name" => $data['name'],
                    "design_category" => $designCategory,
                    "design_perform_species" => $designPerformSpecies,
                ]
            ];

            $entityService->fillBackScriptDetailedListData($detailedListData);
        }
    }

    /**
     * 处理版本事件
     * @param array $versionChangeEvents
     * @return void
     * @throws Exception
     */
    private function handleVersionUpdateEvents(array $versionChangeEvents)
    {
        $versionObServerSubject = observer_subject_by_class_name(VersionModel::class);
        $versionObServerSubject->attach(OutsideVersionUpdateSyncToInsideVersion::getInstance());
        $versionObServerSubject->attach(InsideVersionPassRecordEndTime::getInstance());
        $versionObServerSubject->attach(VersionUploadedRecordUploadedVersion::getInstance());
        $versionObServerSubject->attach(VersionPublishNotifyRelateTask::getInstance());

        $versionList = model(VersionModel::class)
            ->forceMasterDB()
            ->where(['id' => ['IN', implode(',', array_column($versionChangeEvents, 'link_id'))]])
            ->select();
        $versionList = array_column($versionList, null, 'id');
        foreach ($versionChangeEvents as $event) {
            try {
                $id = $event['link_id'];
                if (!isset($versionList[$id])) {
                    continue;
                }
                $versionObServerSubject->setSplPrimaryId($id);
                $versionObServerSubject->setSubjectData($versionList[$id]);
                $versionObServerSubject->splDataChanged($event['record']['data']['old'], $event['record']['data']['new']);
                $versionObServerSubject->notify();
            } catch (Throwable $e) {
                trace(format_exception_info($e), 'ERR');
            }
        }
    }

    /**
     * 处理审批事件
     * @param array $reviewRequestCreateEvent
     * @return void
     */
    private function handleReviewRequestEvents(array $reviewRequestCreateEvent)
    {
        foreach ($reviewRequestCreateEvent as $event) {
            switch ($event['operate']) {
                case 'create':
                    Hook::listen('after_review_request_created', $event);
                    $hookData = [
                        'type' => 'commit_event',
                        'data' => $event,
                    ];
                    Hook::listen('task_review', $hookData);
                    break;
            }
        }
    }

    /**
     * 处理note创建事件
     * @param $createEvents
     * @return void
     */
    private function handleNoteCreateEvents($createEvents)
    {
        foreach ($createEvents as $createEvent) {
            // 已迁移
//            Hook::listen('after_note_create', $createEvent);
            $hookData = [
                'type' => 'note_has_new_comment',
                'data' => $createEvent['record']['data'] ?? [],
            ];
            Hook::listen('note_created_after', $hookData);
        }
    }

    /**
     * 镜头创建 关联资产添加剧集关系
     * @param $data
     * @param $episode
     * @return void
     * @throws Exception
     */
    private function entityCreateRelationEntityAddEpisodeRelation($data, $episode)
    {
        if ($data['module_id'] != module_by_code('shot')['id']
            || empty($episode)
        ) {
            return;
        }
        EntityEpisodeRelationService::getInstance()->syncShotEpisodeRelationToAsset(['project_id' => $data['project_id'], 'id' => $data['id']]);
    }

    /**
     * 处理版本事件
     * @param array $noteChangeEvents
     */
    private function handleNoteUpdateEvents(array $noteChangeEvents)
    {
        $noteIds = [];
        foreach ($noteChangeEvents as $event) {
            $currentOld = $event['record']['data']['old'];
            $currentNew = $event['record']['data']['new'];
            //由等待返修状态到返修中状态
            if (isset($currentOld["demand_status"])) {
                if ($currentOld["demand_status"] == NoteModel::DEMAND_STATUS_PENDING &&
                    $currentNew["demand_status"] == NoteModel::DEMAND_STATUS_PROCESSING) {
                    $noteIds[] = $event['link_id'];
                }
            }
            // 需求变更记录
            Hook::listen('after_note_update', $event);
        }

        try {
            if (!empty($noteIds)) {
                $noteList = model(NoteModel::class)->where(["id" => ["IN", $noteIds]])->field("review_feedback_id")->select();
                $reviewFeedbackIds = array_unique(array_column($noteList, "review_feedback_id"));
                //将未开始的审核反馈状态修改进行中
                $reviewFeedbackList = model(ReviewFeedbackModel::class)
                    ->where(["id" => ["IN", $reviewFeedbackIds], 'status' => ReviewFeedbackModel::STATUS_NOT_STARTED])
                    ->field("id")
                    ->select();

                if (!empty($reviewFeedbackList)) {
                    $reviewFeedbackIdData = array_column($reviewFeedbackList, 'id');
                    model(ReviewFeedbackModel::class)->where(["id" => ['IN', $reviewFeedbackIdData]])->save(['status' => ReviewFeedbackModel::STATUS_IN_PROGRESS]);
                }
            }
        } catch (Throwable $e) {
            trace(format_exception_info($e), 'ERR');
        }
    }

    /**
     * 处理任务事件
     * @param $taskCreateEvent
     * @return void
     * @throws \Exception
     */
    private function handleTaskCreatedEvents($taskCreateEvent)
    {
        foreach ($taskCreateEvent as $event) {
            $data = $event['record']['data'];
            Client::send('calculate_task_man_hour', build_queue_data_with_xu(['task_id' => $data['id']]), 10);
            if (!empty($data['episode_id'])) {
                // 补充 剧集link到tag表中
                model(TaskRelationTagModel::class)->addItem([
                    'tenant_id' => $data['tenant_id'],
                    'task_id' => $data['id'],
                    'link_module_code' => 'episode',
                    'link_id' => $data['episode_id'],
                    'project_id' => $data['project_id'],
                ]);
            }

            if (!empty($data['step_category_id'])) {
                model(TaskRelationTagModel::class)->addItem([
                    'tenant_id' => $data['tenant_id'],
                    'task_id' => $data['id'],
                    'link_module_code' => 'step_category',
                    'link_id' => $data['step_category_id'],
                    'project_id' => $data['project_id'],
                ]);
            }
            //任务上游关系  存储
            Client::send('task_upstream', build_queue_data_with_xu(['task_id' => $data['id']]));
            // 模板待办上下游关系 存储
//            Client::send('task_create_template_todo_fix_upstream_task_ids', build_queue_data_with_xu($data));
        }
    }

    /**
     * 处理工序上游更新时间
     * @param $eventsGrouped
     * @return void
     * @throws Exception
     */
    private function handleStepUpstreamUpdateEvents($eventsGrouped)
    {
        $obServerSubject = observer_subject_by_class_name(StepUpstreamModel::class);
        $obServerSubject->attach(StepUpstreamUpdateRemoveCache::getInstance());

        $stepUpstreamList = model(StepUpstreamModel::class)
            ->where(['id' => ['IN', implode(',', array_column($eventsGrouped, 'link_id'))]])
            ->select();
        $stepUpstreamList = array_column($stepUpstreamList, null, 'id');
        foreach ($eventsGrouped as $event) {
            try {
                $id = $event['link_id'];
                $obServerSubject->setSplPrimaryId($id);
                $obServerSubject->setSubjectData($stepUpstreamList[$id]);
                $obServerSubject->splDataChanged($event['record']['data']['old'], $event['record']['data']['new']);
                $obServerSubject->notify();
            } catch (Throwable $e) {
                trace(format_exception_info($e), 'ERR');
            }
        }
    }

    /**
     * 处理工序上游创建事件
     * @param $createEvents
     * @return void
     */
    private function handleStepUpstreamCreateEvents($createEvents)
    {
        foreach ($createEvents as $event) {
            $data = $event['record']['data'];
            if (!empty($data['project_template_id']) && !empty($data['step_id'])) {
                StepService::getInstance()->removeStepUpstreamQueryCache($data['step_id'], $data['project_template_id']);
            }
        }
    }

    /**
     * 处理版本创建事件
     * @param $createEvents
     * @return void
     */
    private function handleVersionCreateEvents($createEvents)
    {
        foreach ($createEvents as $event) {
            Hook::listen('after_version_created', $event);
        }
    }

    /**
     * 处理审核反馈创建事件
     * @param $createEvents
     */
    public function handleReviewCreateEvents($createEvents)
    {
        foreach ($createEvents as $createEvent) {
            $data = $createEvent['record']['data'];

            if (!empty($data['entry_limit_time'])) {
                if ($data["entry_limit_time"] > time()) {
                    Client::send("change_review_feedback_permission", build_queue_data_with_xu([
                        "change_time" => $data["entry_limit_time"],
                        "data_field" => "entry_limit_time",
                        "review_feedback_id" => $data["id"],
                        "change_to" => ReviewFeedbackModel::PERMISSIONS_TWO,
                    ]), $data["entry_limit_time"] - time());
                }
            }
        }
    }


    /**
     * 处理审核反馈更新事件
     * @param $updateEvents
     */
    public function handleReviewFeedbackUpdateEvents($updateEvents)
    {
        foreach ($updateEvents as $event) {
            $currentNew = $event['record']['data']['new'];
            if (isset($currentNew['entry_limit_time'])) {
                if ($currentNew["entry_limit_time"] > time()) {
                    Client::send("change_review_feedback_permission", build_queue_data_with_xu([
                        "change_time" => $currentNew["entry_limit_time"],
                        "data_field" => "entry_limit_time",
                        "review_feedback_id" => $event['record']["primary_id"],
                        "change_to" => ReviewFeedbackModel::PERMISSIONS_TWO,
                    ]), $currentNew["entry_limit_time"] - time());
                }
            }
        }
    }

    /**
     * 处理工单更新事件
     * @param $eventsGrouped
     * @return void
     * @throws Exception
     */
    private function handleTaskWorkOrderUpdateEvents($eventsGrouped)
    {
        $obServerSubject = observer_subject_by_class_name(TaskWorkOrderModel::class);
        $obServerSubject->attach(StatusChangeSendMessageToChat::getInstance());

        $taskWorkOrderList = model(TaskWorkOrderModel::class)
            ->where(['id' => ['IN', implode(',', array_column($eventsGrouped, 'link_id'))]])
            ->select();
        $taskWorkOrderList = array_column($taskWorkOrderList, null, 'id');
        foreach ($eventsGrouped as $event) {
            try {
                $id = $event['link_id'];
                $obServerSubject->setSplPrimaryId($id);
                $obServerSubject->setSubjectData($taskWorkOrderList[$id]);
                $obServerSubject->splDataChanged($event['record']['data']['old'], $event['record']['data']['new']);
                $obServerSubject->notify();
            } catch (Throwable $e) {
                trace(format_exception_info($e), 'ERR');
            }
        }
    }

    /**
     * 更新原计划
     * @param $eventsGrouped
     * @return void
     */
    private function handleUpdateOriginalPlanNodeEvents($eventsGrouped)
    {
        foreach ($eventsGrouped as $event) {
            Client::send('update_original_plan_node', build_queue_data_with_xu($event));
        }
    }

}
